package com.brycen.nio.multithread;

import sun.nio.ch.SelectionKeyImpl;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * 多线程版本的selector，主要结构如下
 *  boss线程：负责客户端的注册连接工作
 *  worker线程：负责客户端的读写工作，这个worker线程可以有多个，具体可以根据cpu核心数来确定
 *
 *  注：多线程下的channel注册会有问题，如：boss线程负责将新的客户端channel的读写注册到worker线程中
 *      因为worker线程的selector的select已经将selector进行了阻塞，所以boss线程将无法注册事件到selector上
 *  解决方案：学习netty，将注册事件放到work线程中，然后再使用wakeup唤醒select
 *
 *  boss1和worker1为有问题的方案
 *  boss2和worker2是解决上面问题的方案
 */

public class MultiThreadServer {
    public static void main(String[] args) throws IOException {
//        boss1();
        boss2();
    }

    public static void boss2() throws IOException {
        Thread.currentThread().setName("boss");
        ServerSocketChannel ssc = ServerSocketChannel.open();
        ssc.bind(new InetSocketAddress(9999));
        ssc.configureBlocking(false);

        Selector selector = Selector.open();
        ssc.register(selector, SelectionKeyImpl.OP_ACCEPT);

        //这个地方也可以创建一个worker数组，然后下面客户端注册的时候采用轮询方式注册到不同到worker线程中
        Worker2 worker = new Worker2("worker-1");

        //是否有事件触发，没有则阻塞
        while (selector.select()>0){
            Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
            while (iterator.hasNext()){
                SelectionKey sk = iterator.next();
                //判断是否为注册事件
                if (sk.isAcceptable()){
                    SocketChannel channel = ssc.accept();
                    channel.configureBlocking(false);
                    System.out.println("客户端开始注册");
                    /**
                     * 解决方案：将channel.register(worker.selector,SelectionKeyImpl.OP_READ);放到worker线程中运行
                     *          然后使用selector的wakeup唤醒select阻塞即可，让它和selector的select方法在同一个线程运行即可
                     */
                    worker.register(channel);

                    System.out.println("客户端完成注册");
                }
                iterator.remove();
            }
        }
    }
    static class Worker2 implements Runnable{
        private Thread thread;
        private Selector selector;
        private String name;
        private volatile boolean isCreate = false;
        //存放客户端注册事件的代码，让它移植到worker线程中去注册
        private ConcurrentLinkedQueue<Runnable> queue = new ConcurrentLinkedQueue();

        public Worker2(String name) {
            this.name = name;
        }

        public void register(SocketChannel channel) throws IOException {
            //确保一个worker线程只有一个选择器
            if (!isCreate){
                selector = Selector.open();
                thread = new Thread(this,name);
                thread.start();
                isCreate = true;
            }
            //当前代码运行的线程还是boss，所以要将客户端注册事件的命令下放到worker线程中执行
            queue.add(()->{
                try {
                    channel.register(selector,SelectionKeyImpl.OP_READ);
                } catch (ClosedChannelException e) {
                    throw new RuntimeException(e);
                }
            });
            //唤醒选择器到select阻塞方法，然后就会执行queue的poll方法拿到注册事件任务
            selector.wakeup();

        }

        @Override
        public void run() {
            while (true){
                try {
                    //是否有事件触发，没有则阻塞
                    selector.select();
                    Runnable runnable = queue.poll();
                    if (runnable!=null){
                        //运行客户端注册事件命令
                        runnable.run();
                    }
                    //获取事件迭代器
                    Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                    while (iterator.hasNext()){
                        SelectionKey sk = iterator.next();
                        try{
                            if (sk.isReadable()){
                                SocketChannel socketChannel = (SocketChannel) sk.channel();
                                ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
                                int read = socketChannel.read(byteBuffer);
                                if (read>0){
                                    byteBuffer.flip();
                                    System.out.println(new String(byteBuffer.array(),0,byteBuffer.remaining()));
                                }
                                //客户端断开连接
                                if (read==-1){
                                    sk.cancel();
                                }
                            }
                        }catch (Exception e){
                            //客户端异常断开连接
                            e.printStackTrace();
                            sk.cancel();
                        }
                        iterator.remove();
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    //===============================================================

    public static void boss1() throws IOException {
        Thread.currentThread().setName("boss");
        ServerSocketChannel ssc = ServerSocketChannel.open();
        ssc.bind(new InetSocketAddress(9999));
        ssc.configureBlocking(false);

        Selector selector = Selector.open();
        ssc.register(selector, SelectionKeyImpl.OP_ACCEPT);

        Worker1 worker = new Worker1("worker-1");
        /**
         * 问题所在1：这个地方的register会使worker线程运行，然后到select进行阻塞
         *          worker.register()这行代码放到"问题所在2"的前面依然也会存在同样的问题
         */
        worker.register();
        //是否有事件触发，没有则阻塞
        while (selector.select()>0){
            Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
            while (iterator.hasNext()){
                SelectionKey sk = iterator.next();
                //判断是否为注册事件
                if (sk.isAcceptable()){
                    SocketChannel channel = ssc.accept();
                    channel.configureBlocking(false);
                    System.out.println("客户端开始注册");
                    /**
                     * 问题所在2：这个地方将channel注册到worker线程到selector选择器上，此时运行到线程上boss线程
                     *          而我们到selector已经在worker线程中阻塞住了，所以下面到代码会一直等待知道worker
                     *          线程中到select有事件触发了并且cpu时间片给到boss才能将客户端到channel读事件注册上
                     */
                    channel.register(worker.selector,SelectionKeyImpl.OP_READ);
                    System.out.println("客户端完成注册");
                }
                iterator.remove();
            }
        }
    }
    static class Worker1 implements Runnable{
        private Thread thread;
        private Selector selector;
        private String name;
        private volatile boolean isCreate = false;

        public Worker1(String name) {
            this.name = name;
        }

        public void register() throws IOException {
            //确保一个worker线程只有一个选择器
            if (!isCreate){
                selector = Selector.open();
                thread = new Thread(this,name);
                thread.start();
                isCreate = true;
            }
        }

        @Override
        public void run() {
            while (true){
                try {
                    //是否有事件触发，没有则阻塞
                    selector.select();

                    //获取事件迭代器
                    Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                    while (iterator.hasNext()){
                        SelectionKey sk = iterator.next();
                        try{
                            if (sk.isReadable()){
                                SocketChannel socketChannel = (SocketChannel) sk.channel();
                                ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
                                int read = socketChannel.read(byteBuffer);
                                if (read>0){
                                    byteBuffer.flip();
                                    System.out.println(new String(byteBuffer.array(),0,byteBuffer.remaining()));
                                }
                                //客户端断开连接
                                if (read==-1){
                                    sk.cancel();
                                }
                            }
                        }catch (Exception e){
                            //客户端异常断开连接
                            e.printStackTrace();
                            sk.cancel();
                        }
                        iterator.remove();
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
